Skip to Content

07. LangChain 与 LangGraph 的工程实践

本章基于 LangChain / LangGraph 官方 TypeScript 文档的当前状态（2026-04-22）组织。 API 演进较快，生产落地时请直接对照 docs.langchain.com  的对应章节。

本章高频面试题

  1. LangChain 和 LangGraph 分别是什么？它们是什么关系？
  2. 什么时候用 LangChain，什么时候用 LangGraph？
  3. 为什么说 LangChain agent 是构建在 LangGraph 之上的？
  4. LangGraph 适合解决哪些普通 workflow 不好解决的问题？
  5. LangChain 的 middleware 系统怎么用？wrapModelCall / wrapToolCall / beforeModel / afterModel 各解决什么问题？
  6. LangGraph 的 state、checkpointer、streaming、interrupts、time travel 分别是什么意思？
  7. 如何用 TypeScript 快速起一个 LangChain agent？
  8. 如何用 LangGraph 构建更强控制力的 agent runtime？
  9. LangGraph 的 interrupt() / Command(resume=...) 模式怎么用？
  10. LangGraph Platform / Server 在生产中扮演什么角色？
  11. 前端如何接 LangGraph 的执行流？useStream 提供什么？

1. LangChain 是什么

LangChain 是一个开源框架，用来帮助开发者更快地构建基于 LLM 的应用和 Agent。

根据当前官方文档，LangChain 的定位是：

  • 提供统一的模型接口
  • 提供工具、消息、短期记忆、流式输出、结构化输出等核心组件
  • 提供较高层的 agent abstraction（createAgent），帮你快速搭建 Agent
  • 通过 middleware 系统支持 context engineering 的精细控制

官方文档里有一个很值得记住的结论：

如果你想快速构建自定义 Agent，从 LangChain 开始。

2. LangGraph 是什么

LangGraph 是一个更低层的 agent orchestration framework 和 runtime。 它更关注：

  • 长时间运行
  • 有状态执行
  • 持久化（checkpointer）
  • durable execution
  • interrupts / human-in-the-loop
  • streaming
  • time travel（状态历史回溯）

官方文档定位非常明确：

它是偏底层、偏编排、偏运行时的框架，适合需要高度控制和复杂工作流/Agent 混合编排的场景。

3. LangChain 和 LangGraph 的关系

官方文档明确指出：

LangChain 的 agents 是构建在 LangGraph 之上的。

这句话非常重要——它解释了为什么 LangChain agent 天然具备 durable execution、streaming、human-in-the-loop、persistence。

  • LangChain 给你更高层的开发体验（createAgent + middleware）
  • LangGraph 给你更底层的执行能力（StateGraph、checkpointer、interrupt）

4. 怎么选

优先 LangChain 的场景

  • 快速起一个 Agent，不想一开始就设计完整 graph
  • 重点在工具、上下文和 prompt 层
  • Middleware 能满足所有定制需求

优先 LangGraph 的场景

  • 需要自定义节点和边的执行逻辑
  • 需要精确控制 state schema 和 reducers
  • 有长时间运行任务（分钟到小时到天级）
  • 需要 interrupt / resume
  • 需要 deterministic workflow 和 agentic workflow 混合
  • 需要 time travel / branching / forking

5. LangChain 现在的核心组件

从最新官方 TypeScript 文档看，LangChain 已经围绕 Agent 提供了一整套高层组件：

  • agents（createAgent）
  • tools（tool() helper）
  • messages
  • short-term memory
  • streaming
  • structured output（responseFormat）
  • middleware
  • context engineering
  • human-in-the-loop
  • retrieval
  • long-term memory

如果还把 LangChain 理解成老式 chains，那已经落后于当前官方主线。

6. LangChain 的 createAgent

当前的核心 API 签名（简化版）：

import { createAgent, tool } from "langchain";
import { z } from "zod";
 
const getWeather = tool(
  async ({ city }) => {
    return `Weather in ${city}: Sunny, 26C`;
  },
  {
    name: "get_weather",
    description:
      "获取指定城市的天气信息。仅用于天气查询；不要用于预报未来或其他城市关联问题。",
    schema: z.object({
      city: z.string().describe("需要查询天气的城市名"),
    }),
  }
);
 
const agent = createAgent({
  model: "anthropic:claude-opus-4-7",
  tools: [getWeather],
  systemPrompt: "你是一个可靠的天气助手。",
  // 可选:
  // responseFormat: z.object({...}),   // 强制结构化输出
  // stateSchema: CustomState,          // 自定义 state
  // middleware: [...],                 // 注入 middleware
});
 
const result = await agent.invoke({
  messages: [{ role: "user", content: "帮我查一下东京天气" }],
});

关键参数：

  • model：支持 provider:model 字符串或直接传 ChatModel 实例
  • tools：工具数组
  • systemPrompt：静态 system prompt
  • responseFormat：传 Zod schema 会走 structured output
  • stateSchema：自定义 state（通常沿用默认）
  • middleware：在 Agent 生命周期注入定制逻辑

7. Middleware：LangChain 做 Context Engineering 的主通道

Middleware 系统是当前 LangChain 最重要的能力之一。它允许在 Agent 生命周期的关键节点拦截和修改行为：

  • beforeModel / afterModel：模型调用前后做消息裁剪、state 更新
  • wrapModelCall：完全包裹模型调用（重试、fallback、成本追踪）
  • wrapToolCall：包裹工具调用（校验、权限、审批门禁）
  • dynamicSystemPromptMiddleware：按 state / runtime context 动态生成 system prompt

7.1 动态 system prompt

import { z } from "zod";
import {
  createAgent,
  dynamicSystemPromptMiddleware,
} from "langchain";
 
const contextSchema = z.object({
  userRole: z.enum(["viewer", "operator", "admin"]),
  workspaceId: z.string(),
});
type RuntimeContext = z.infer<typeof contextSchema>;
 
const agent = createAgent({
  model: "anthropic:claude-opus-4-7",
  tools: [],
  contextSchema,
  middleware: [
    dynamicSystemPromptMiddleware<RuntimeContext>((state, runtime) => {
      const parts = ["你是一个可靠的企业助手。"];
      if (runtime.context.userRole === "viewer") {
        parts.push("当前用户只读,禁止建议执行写操作。");
      }
      if (state.messages.length > 10) {
        parts.push("当前对话较长,请保持回答简洁,只突出关键结论。");
      }
      return parts.join("\n");
    }),
  ],
});

7.2 用 createMiddleware 自定义

import { createMiddleware } from "langchain";
 
const budgetGuard = createMiddleware({
  name: "budget-guard",
  beforeModel: async ({ state }) => {
    const used = state.messages.reduce(
      (sum, m) => sum + (m.usage_metadata?.total_tokens ?? 0),
      0
    );
    if (used > 100_000) {
      throw new Error("token_budget_exhausted");
    }
  },
  wrapToolCall: async (toolCall, next) => {
    if (HIGH_RISK_TOOLS.has(toolCall.name)) {
      await requireApproval(toolCall);
    }
    return next(toolCall);
  },
});

这种组合方式让你能把权限、预算、审批、日志、评估 hook 作为可组合的 middleware，而不是塞进 agent 主流程。

8. LangGraph 的核心概念

8.1 State

State 是整个 graph 在执行过程中共享和更新的数据结构。当前 TypeScript API 用 StateSchema 配合 Zod：

import {
  StateSchema,
  MessagesValue,
  ReducedValue,
  StateGraph,
  START,
  END,
} from "@langchain/langgraph";
import { z } from "zod";
 
const State = new StateSchema({
  messages: MessagesValue,            // 带默认 appending reducer
  llmCalls: new ReducedValue(
    z.number().default(0),
    { reducer: (x, y) => x + y }      // 自定义累加 reducer
  ),
});
  • MessagesValue：内建 reducer，自动追加消息（不覆盖）
  • ReducedValue：自定义 reducer，用来累加、合并、去重

8.2 Node

Node 是一个执行单元：调模型、调工具、做判断、做转换。

8.3 Edge

Edge 定义节点之间的执行流向，表达线性流、条件分支、回环、子图调用。

8.4 Checkpointer

持久化执行状态，允许任务中断后恢复。主要实现：

  • MemorySaver：内存，用于开发和测试
  • PostgresSaver：生产首选（@langchain/langgraph-checkpoint-postgres）
  • RedisSaver：高频 checkpoint 场景
  • SQLiteSaver：本地开发/单机场景

Checkpointer 才是 LangGraph “durable execution” 的核心——所有状态都落到 checkpointer，任务中断/重启都能恢复。

8.5 Interrupts

在图执行过程中主动中断，让人或外部系统介入，然后继续。当前推荐的是 interrupt() + Command(resume=...) 模式（见 §11）。

8.6 Streaming

把节点状态、tokens、custom events、tool events 持续暴露出来。支持多种 stream mode（见 §10）。

8.7 Time travel

所有 checkpoint 都保留，允许从历史任意一点”分叉”出新分支。这对调试、A/B 不同决策、回滚都很有用。

9. 极简 LangGraph 示例

import {
  StateSchema,
  MessagesValue,
  StateGraph,
  START,
  END,
} from "@langchain/langgraph";
import { MemorySaver } from "@langchain/langgraph";
import { ChatAnthropic } from "@langchain/anthropic";
 
const State = new StateSchema({
  messages: MessagesValue,
});
 
const model = new ChatAnthropic({ model: "claude-opus-4-7" });
 
const graph = new StateGraph(State)
  .addNode("llmCall", async (state) => {
    const response = await model.invoke(state.messages);
    return { messages: [response] };
  })
  .addEdge(START, "llmCall")
  .addEdge("llmCall", END)
  .compile({ checkpointer: new MemorySaver() });
 
const result = await graph.invoke(
  { messages: [{ role: "user", content: "Hello" }] },
  { configurable: { thread_id: "thread_123" } }
);

关键点：

  • 节点返回 state 增量（{ messages: [response] }），不是完整 state
  • compile() 时注入 checkpointer，thread_id 绑定会话
  • 同一 thread_id 再次 invoke 会基于上次 checkpoint 继续

10. LangGraph Streaming

当前（2026-04）LangGraph JS 官方支持的 stream mode：

  • values：每步后的完整 state
  • updates：每步的 state 增量
  • messages：LLM token 流（2-tuple [token, metadata]）
  • custom：节点内通过 writer 发出的自定义事件
  • tools：工具生命周期（on_tool_start / on_tool_event / on_tool_end / on_tool_error）
  • debug：完整执行信息（调试用）

10.1 单 mode

for await (const chunk of await graph.stream(
  { messages: [{ role: "user", content: "Hi" }] },
  { streamMode: "updates", configurable: { thread_id: "t1" } }
)) {
  console.log(chunk);
}

10.2 多 mode

for await (const [mode, chunk] of await graph.stream(inputs, {
  streamMode: ["updates", "custom"],
  configurable: { thread_id: "t1" },
})) {
  console.log(mode, chunk);
}

10.3 节点内发 custom 事件

.addNode("research", async (state, config) => {
  config.writer({ phase: "search", progress: 0.3 });
  const results = await search(state.query);
  config.writer({ phase: "rank", progress: 0.8 });
  return { results };
})

这让你能把”过程性进度”和”状态更新”分开推送，前端可以分别渲染。

11. Interrupt / Resume：HITL 的一等公民

当前推荐模式（2025 改造后）：

11.1 在节点内暂停

import { interrupt } from "@langchain/langgraph";
 
.addNode("approval", async (state) => {
  const decision = interrupt({
    type: "refund_approval",
    amount: state.refundAmount,
    reason: state.reason,
  });
  // 执行到这里会抛出 GraphInterrupt,runtime 把 payload 写入 checkpoint
  // 下次带 Command(resume=...) invoke 时,从此节点重新开始,
  // interrupt() 调用会直接返回 resume 传入的值
  return {
    approved: decision.approved,
    approverId: decision.approverId,
  };
})

11.2 外部恢复

import { Command } from "@langchain/langgraph";
 
// 第一次 invoke:执行到 interrupt 时返回,状态持久化
const first = await graph.invoke(input, {
  configurable: { thread_id: "t1" },
});
// first["__interrupt__"] 里是 interrupt payload
 
// 人工审批后恢复(可以是数月后、另一台机器):
await graph.invoke(
  new Command({ resume: { approved: true, approverId: "u123" } }),
  { configurable: { thread_id: "t1" } }
);

11.3 关键规则

  • Node 会从头重新执行：resume 后 interrupt 之前的代码会再跑一次。有副作用的动作要么放在 interrupt 之后，要么做幂等
  • 不要条件性跳过 interrupt：同一节点内 interrupt 的顺序和数量必须确定，否则 resume 匹配会错位
  • Command(resume=…) 只能作为 invoke/stream 的输入，其他 Command 参数（update、goto、graph）是节点返回值用的
  • 结构化 resume 值：传 dict / list 而不是纯字符串，方便多字段场景

这个模式配合 Postgres checkpointer 能实现”暂停几个月后在另一台机器上恢复”的能力，对需要人工审批、多天长任务的业务非常关键。

12. LangGraph Platform / Server 在生产中的角色

LangGraph 不只是一个库，还有 LangGraph Platform（托管）/ LangGraph Server（自托管）：

  • API server：把你的 graph 暴露成 REST + SSE endpoint
  • Thread management：自动管理 thread、checkpoint、history
  • Worker pool：长任务后台执行
  • Scheduling：cron 触发 graph run
  • Deployment：langgraph deploy 一键上线
  • UI：LangGraph Studio（本地 IDE 式可视化调试）

什么时候值得用：

  • Agent 需要作为独立后端服务暴露给多个前端
  • 需要后台长任务 + scheduling
  • 想省掉自己搭 SSE + thread store + worker 的功夫

如果你已经有成熟的 backend 框架（NestJS、Next.js），也可以只用 LangGraph 作为库，不上 Platform。

13. 前端接入：useStream

import { useStream } from "@langchain/langgraph-sdk/react";
 
export function Chat() {
  const thread = useStream({
    apiUrl: process.env.NEXT_PUBLIC_LANGGRAPH_URL!,
    assistantId: "agent",
    onUpdateEvent: (update) => {
      console.log("graph update", update);
    },
    onError: (error) => {
      console.error(error);
    },
  });
 
  return (
    <div>
      {thread.messages.map((message, idx) => (
        <div key={message.id ?? idx}>{String(message.content ?? "")}</div>
      ))}
      {thread.interrupt && (
        <ApprovalCard
          payload={thread.interrupt.value}
          onDecide={(decision) => thread.submit(undefined, { command: { resume: decision } })}
        />
      )}
    </div>
  );
}

useStream 自动处理：

  • SSE 连接和重连
  • Thread ID 管理
  • Interrupt 状态暴露（thread.interrupt）
  • Resume 提交（thread.submit(undefined, { command: { resume } })）
  • 消息流增量合并

14. LangChain / LangGraph / Deep Agents 怎么理解

根据当前官方表述：

  • LangChain：高层 Agent 开发框架（createAgent + middleware），快速起 agent 首选
  • LangGraph：底层 runtime（state、checkpointer、interrupt、streaming），精确控制首选
  • Deep Agents：预构建的高级 agent 模式（planning tool + file system + sub-agents），长任务场景首选
  • LangGraph Platform / Server：把 graph 部署成服务

这不是互斥关系——Deep Agents 本身基于 LangGraph 构建，LangChain Agents 也构建在 LangGraph 上。

15. 面试总结

当前 LangChain 是高层 Agent 开发框架，提供 createAgent、tools、messages、middleware、memory、retrieval；LangGraph 是底层 runtime 和 orchestration 框架，提供 state、checkpointer、durable execution、interrupt/resume、streaming、time travel。官方明确说 LangChain agents 构建在 LangGraph 之上。我的选择是：简单 agent 用 createAgent + middleware，需要精确控制和长时任务直接上 LangGraph，需要长任务规划用 Deep Agents 模式。生产部署考虑 LangGraph Server 或 Platform 承载 thread 管理、checkpoint 持久化和 SSE。前端用 useStream 直接接 graph，interrupt / resume 天然支持 HITL。

16. 本章方法论小结

  1. 当前 LangChain 已经是高层 Agent 框架，不是老式 chains 库
  2. LangGraph 是底层 state machine + durable runtime
  3. LangChain agents 构建在 LangGraph 之上，两者是分层关系不是竞争关系
  4. Middleware 是 LangChain 做 Context Engineering 的主通道（beforeModel/afterModel/wrapModelCall/wrapToolCall/dynamicSystemPromptMiddleware）
  5. LangGraph 的 state + checkpointer + interrupt + streaming 是生产 Agent runtime 的一等公民
  6. interrupt() + Command(resume=...) 让 HITL 变得可恢复（含跨机器、跨天）
  7. LangGraph Platform/Server 把 graph 变成可部署的服务，省掉很多胶水代码
  8. 前端用 useStream 能直接消费 graph 的事件流和 interrupt 状态
Last updated on
从 Copilot 到 Agent